package com.g.ocache.impl.store;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;

import com.g.ocache.Cache;
import com.g.ocache.ValidationException;


/**
 * 
 * @author Jingqi Xu
 */
public abstract class AbstractConcurrentStore<K, V> extends AbstractStore<K, V> {
	//
	protected static final int DEFAULT_CONSURRENCY_LEVEL = 16;
	protected static final int DEFAULT_INITIAL_CAPACITY = 128;
	protected static final float DEFAULT_GROWTH_FACTOR = 0.75F;
	
	//
	private final int segmentMask;
	private final int segmentShift;
	private final int segmentCount;
	private final int concurrencyLevel;
	private List<Bucket<K, V>> buckets;
	private final ReentrantReadWriteLock lock;
	
	//
	protected abstract List<Bucket<K, V>> createBuckets(int count);
	
	
	/**
	 * 
	 */
	public AbstractConcurrentStore(Cache<K, V> cache) {
		this(cache, DEFAULT_CONSURRENCY_LEVEL, false);
	}
	
	public AbstractConcurrentStore(Cache<K, V> cache, int concurrencyLevel) {
		this(cache, concurrencyLevel, false);
	}
	
	public AbstractConcurrentStore(Cache<K, V> cache, int concurrencyLevel, boolean fair) {
		//
		super(cache);
		
		//
		this.concurrencyLevel = concurrencyLevel;
		int sshift = 0;
		int ssize = 1;
		while (ssize < this.concurrencyLevel) {
			++sshift;
			ssize <<= 1;
		}
		this.segmentShift = 32 - sshift;
		this.segmentMask = ssize - 1;
		this.segmentCount = ssize;
		
		//
		this.lock = new ReentrantReadWriteLock(fair);
	}
	
	/**
	 * 
	 */
	public String toString() {
		return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
		.append("cache", getCache().getName())
		.append("concurrencyLevel", concurrencyLevel)
		.append("running", isRunning()).toString();
	}
	
	/**
	 * 
	 */
	protected void doStart() throws Exception {
		this.buckets = createBuckets(segmentCount);
	}
	
	protected void doStop() throws Exception {
		// NOP
	}
	
	/**
	 * 
	 */
	public final boolean isFair() {
		return lock.isFair();
	}
	
	public final int getConcurrencyLevel() {
		return concurrencyLevel;
	}
	
	public void lock(boolean exclusive) {
		final Lock l = exclusive ? this.lock.writeLock() : this.lock.readLock();
		l.lock();
	}
	
	public void unlock(boolean exclusive) {
		final Lock l = exclusive ? this.lock.writeLock() : this.lock.readLock();
		l.unlock();
	}
	
	/**
	 * 
	 */
	public final int size() {
		int r = 0;
		lock(true); // Lock all buckets
		try {
			for (Bucket<K, V> bucket : buckets) {
				r += bucket.size();
			}
		} finally {
			unlock(true);
		}
		return r;
	}
	
	public final boolean containsKey(K key) {
		lock(false);
		try {
			final Bucket<K, V> bucket = getBucket(key);
			return bucket.containsKey(key);
		} finally {
			unlock(false);
		}
	}
	
	public final void clear() {
		lock(true); // Lock all buckets
		try {
			for(Bucket<K, V> bucket : buckets) {
				bucket.clear();
			}
		} finally {
			unlock(true);
		}
	}
	
	public final V remove(K key) {
		lock(false);
		try {
			final Bucket<K, V> bucket = getBucket(key);
			return bucket.remove(key);
		} finally {
			unlock(false);
		}
	}
	
	public final V get(K key) {
		lock(false);
		try {
			final Bucket<K, V> bucket = getBucket(key);
			return bucket.get(key);
		} finally {
			unlock(false);
		}
	}
	
	public final Map<K, V> getAll() {
		Map<K, V> r = new HashMap<K, V>();
		lock(true); // Lock all buckets
		try {
			for(Bucket<K, V> bucket : buckets) {
				r.putAll(bucket.getAll());
			}
		} finally {
			unlock(true);
		}
		return r;
	}

	public final V put(K key, V value) {
		lock(false);
		try {
			final Bucket<K, V> bucket = getBucket(key);
			return bucket.put(key, value);
		} finally {
			unlock(false);
		}
	}

	public final V putIfAbsent(K key, V value) {
		lock(false);
		try {
			final Bucket<K, V> bucket = getBucket(key);
			return bucket.putIfAbsent(key, value);
		} finally {
			unlock(false);
		}
	}

	public final Map<K, V> putAll(Map<K, V> values) {
		Map<K, V> r = new HashMap<K, V>(values.size());
		lock(true); // Lock all buckets
		try {
			for(K key : values.keySet()) {
				Bucket<K, V> bucket = getBucket(key);
				try {
					//
					V existing = bucket.put(key, values.get(key));
					
					//
					if(existing != null) {
						r.put(key, existing);
					}
				} catch(ValidationException ve) {
					// If validator threw an exception, that exception can be found in "Reject" event
				} 
			}
		} finally {
			unlock(true);
		}
		return r;
	}
	
	/**
	 * 
	 */
	protected final List<Bucket<K, V>> getBuckets() {
		return buckets;
	}
	protected final Bucket<K, V> getBucket(K key) {
		final int segment = (hash(key) >>> segmentShift) & segmentMask;
		return buckets.get(segment);
	}
	
	private static final int hash(Object key) {
		int h = key.hashCode();
        h += (h <<  15) ^ 0xffffcd7d;
        h ^= (h >>> 10);
        h += (h <<   3);
        h ^= (h >>>  6);
        h += (h <<   2) + (h << 14);
        return h ^ (h >>> 16);
	}
	
	/**
	 * 
	 */
	protected interface Bucket<K, V> {
		
		int size();
		
		boolean containsKey(K key);
		
		void clear();

		V remove(K key);
		
		V get(K key);
		
		Map<K, V> getAll();
		
		V put(K key, V value);
		
		V putIfAbsent(K key, V value);
	}
}
